package tmt.usercenter.web.service;

import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteSemaphore;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import tmt.usercenter.web.bean.TmtSemaphore;
import tmt.usercenter.web.domain.BaseDomain;

import java.util.*;
import java.util.concurrent.Semaphore;
import java.util.stream.Collectors;

/**
 * 系统中的全局信号量服务。
 */
@Component
public class SemaphoreService {

    private final static String SEMAPHORE_NAME_FORMAT = "Semaphore_%s_%s";

    @Autowired(required = false)
    private Ignite igniteInstance;

    /**
     * 获取信号量对象。
     *
     * @param clazz 类型
     * @param id    对象ID
     * @return
     */
    public TmtSemaphore getSemaphore(Class<? extends BaseDomain> clazz, Long id) {
        String semName = String.format(SEMAPHORE_NAME_FORMAT, clazz.getSimpleName(), id);
        return getSemaphoreByName(semName);
    }

    public List<TmtSemaphore> acquires(Class<? extends BaseDomain> clazz, List<Long> ids) {
        List<TmtSemaphore> list = new ArrayList<>();

        if (ids.size() <= 0)
            return list;

        ids.stream().distinct().sorted(Comparator.naturalOrder())
                .collect(Collectors.toList())
                .forEach(x -> {
                    TmtSemaphore semaphore = getSemaphore(clazz, x);
                    semaphore.acquire();
                    list.add(semaphore);
                });
        return list;
    }

    public List<TmtSemaphore> acquires(List<? extends BaseDomain> objectList) {
        if (objectList.size() <= 0)
            return new ArrayList<>();

        Class<? extends BaseDomain> clazz = objectList.get(0).getClass();
        List<Long> ids = objectList.stream().map(BaseDomain::getId)
                .collect(Collectors.toList());

        return acquires(clazz, ids);
    }

    public void releases(List<TmtSemaphore> semaphores) {
        if (CollectionUtils.isEmpty(semaphores))
            return;

        Collections.reverse(semaphores);
        semaphores.forEach(TmtSemaphore::release);
    }


    private TmtSemaphore getSemaphoreByName(String semName) {
        TmtSemaphore semaphore = null;
        synchronized (TmtSemaphore.LocalSemaphoreMap) {
            semaphore = TmtSemaphore.LocalSemaphoreMap.get(semName);
            if (semaphore == null) {
                semaphore = igniteInstance == null ? new TmtSemaphore(semName, new Semaphore(1))
                        : new TmtSemaphore(semName, igniteInstance.semaphore(
                        semName, // Distributed semaphore name.
                        1,        // Number of permits.
                        true,      // Release acquired permits if node, that owned them, left topology.
                        true       // Create if it doesn't exist.
                ));
                TmtSemaphore.LocalSemaphoreMap.put(semName, semaphore);
            }
        }
        return semaphore;
    }

}
